package com.slipper.collect;

import com.slipper.PaasContext;
import com.slipper.system.collect.bo.CollectTaskBO;
import com.slipper.system.collect.bo.CollectTaskLogBO;
import com.slipper.system.collect.service.CollectTaskLogService;
import com.slipper.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class CollectTask implements Runnable {
    private final Logger logger = LoggerFactory.getLogger(CollectTask.class);

    private final CollectTaskBO collectTask;
    private CollectTaskLogBO collectTaskLog;
    private ExecutorService service;
    private final Map<String, Future<?>> futureMap;
    private final Map<String, CollectTable> collectTableMap;

    public CollectTask(CollectTaskBO collectTask) {
        this.collectTask = collectTask;
        this.futureMap = new HashMap<>();
        this.collectTableMap = new HashMap<>();
    }

    @Override
    public void run() {
        collectTaskLog = createLog(collectTask);
        PaasContext.getBean(CollectTaskLogService.class).insert(collectTaskLog);

        int threadCount = 10;
        if (collectTask.getSyncCount() != null && collectTask.getSyncCount() > 0) {
            threadCount = collectTask.getSyncCount();
        }

        service = Executors.newFixedThreadPool(threadCount);
        try {
            collectTask.getTableList().forEach(item -> {
                CollectTable collectTable = new CollectTable(collectTaskLog, item);
                futureMap.put(item.getId(),  service.submit(collectTable));
                collectTableMap.put(item.getId(), collectTable);
            });
            service.shutdown();
            if (!service.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS)) {
                logger.warn("等待超时！");
            }
        } catch (Exception e) {
            logger.error("任务执行失败！", e);
        } finally {
            collectTaskLog.setExecuteStatus("执行成功");
            collectTableMap.values().forEach(item -> {
                collectTaskLog.setAddCount(collectTaskLog.getAddCount() + item.getAddCount());
                collectTaskLog.setUpdateCount(collectTaskLog.getUpdateCount() + item.getUpdateCount());
                if (!item.isSuccess()) {
                    collectTaskLog.setExecuteStatus("执行失败");
                }
            });
            collectTaskLog.setEndTime(System.currentTimeMillis());
            collectTaskLog.setExecuteTime(collectTaskLog.getEndTime() - collectTaskLog.getStartTime());
            PaasContext.getBean(CollectTaskLogService.class).update(collectTaskLog);
        }
    }

    public void stopTask(String taskId) {
        collectTask.getTableList().forEach(item -> stopTableTask(item.getId()));
    }

    public void stopTableTask(String tableId) {
        CollectTable collectTable = collectTableMap.get(tableId);
        if (collectTable != null) {
            collectTable.shutdown();
        }

        Future<?> future = futureMap.get(tableId);
        if (future != null) {
            future.cancel(true);
        }
    }

    private CollectTaskLogBO createLog(CollectTaskBO collectTask) {
        CollectTaskLogBO collectTaskLog = new CollectTaskLogBO();
        collectTaskLog.setId(UUID.getUUID());
        collectTaskLog.setTaskId(collectTask.getId());
        collectTaskLog.setTaskName(collectTask.getName());
        collectTaskLog.setOriginDatasourceId(collectTask.getOriginDatasourceId());
        collectTaskLog.setTargetDatasourceId(collectTask.getTargetDatasourceId());
        collectTaskLog.setStartTime(System.currentTimeMillis());
        collectTaskLog.setExecuteStatus("执行中");
        collectTaskLog.setAddCount(0L);
        collectTaskLog.setUpdateCount(0L);
        return collectTaskLog;
    }
}
